package cn.jly.bigdata.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author lanyangji
 * @date 2019/11/25 22:02
 */
object Spark03_Serializable {

  def main(args: Array[String]): Unit = {

    val sc: SparkContext = new SparkContext(new SparkConf().setMaster("local[*]").setAppName("test-01"))

    val arrayRDD: RDD[String] = sc.makeRDD(Array("hello", "spark", "hive", "lanyangji"))

    // 创建查询对象
    val search = new Search("el")

    val conRDD: RDD[String] = search.getContainRDD2(arrayRDD)
    conRDD.foreach(println)
  }
}


// 自定义查询对象
// query没加val和var修饰符，如果类中方法没有用到，则是局部变量，如果方法用到了（属性初始化不算)，则会变成final的私有属性
// 看ScalaTest.java的反编译
class Search(queryStr: String) extends Serializable {

  // 是否包含查询子串
  def isContain(s: String): Boolean = s.contains(queryStr)

  // 过滤出包含查询字符串的RDD
  def getContainRDD(rdd: RDD[String]): RDD[String] = rdd.filter(isContain)

  // 过滤出包含查询字符串的RDD
  def getContainRDD2(rdd: RDD[String]): RDD[String] = {
    // 除了实现序列化接口，这种方法也可以避免报出未序列化异常
    //val query_ = this.queryStr
    //rdd.filter(_.contains(query_))

    rdd.filter(_.contains(queryStr))
  }
}